

import os
import time
import random
import json
import uuid
from faker import Faker
from kafka import KafkaProducer, errors

fake = Faker('zh_CN')

# 1. 从环境变量读取 bootstrap，拆分成列表
bootstrap = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'kafka:9092')
brokers = [b.strip() for b in bootstrap.split(',') if b.strip()]

# 2. 循环重试连接 Kafka
while True:
    try:
        producer = KafkaProducer(
            bootstrap_servers=brokers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            api_version_auto_timeout_ms=30000  # 可选：延长自动探测超时
        )
        print(f"[INFO] 成功连接 Kafka brokers: {brokers}")
        break
    except errors.NoBrokersAvailable:
        print(f"[WARN] Kafka brokers {brokers} 暂不可用，1秒后重试…")
        time.sleep(1)

# 3. 数据生成函数
def gen_user():
    return {
        'user_id': str(uuid.uuid4()),
        'name': fake.name(),
        'gender': random.choice(['男', '女']),
        'age': random.randint(18, 60),
        'email': fake.email(),
        'address': fake.address(),
    }

def gen_product():
    return {
        'product_id': str(uuid.uuid4()),
        'product_name': fake.word(),
        'category': random.choice(['电子', '服装', '家居']),
        'price': round(random.uniform(10, 5000), 2),
        'stock': random.randint(0, 1000),
    }

def gen_seller():
    return {
        'seller_id': str(uuid.uuid4()),
        'seller_name': fake.company(),
        'location': fake.city_name(),
    }

def gen_order(user_id, product_id):
    return {
        'order_id': str(uuid.uuid4()),
        'user_id': user_id,
        'product_id': product_id,
        'amount': round(random.uniform(20, 2000), 2),
        'qty': random.randint(1, 5),
        'ts': fake.date_time().isoformat(),  # 注意这里要调用 fake.date_time()
    }

# 4. 生产并发送到 Kafka
if __name__ == '__main__':
    # 先生成一些静态的用户／产品／商家列表
    users = [gen_user() for _ in range(10)]
    products = [gen_product() for _ in range(20)]
    sellers = [gen_seller() for _ in range(5)]

    # 将用户/产品/商家等发到对应主题
    for u in users:
        producer.send('users', u)
    for p in products:
        producer.send('products', p)
    for s in sellers:
        producer.send('sellers', s)
    producer.flush()
    print(f"[INFO] 已发送 {len(users)} users, {len(products)} products, {len(sellers)} sellers")

    # 然后不停地发送订单
    try:
        while True:
            u = random.choice(users)['user_id']
            p = random.choice(products)['product_id']
            order = gen_order(u, p)
            producer.send('orders', order)
            # 每隔 0.5~2 秒发送一条
            time.sleep(random.uniform(0.5, 2.0))
    except KeyboardInterrupt:
        print("[INFO] 退出数据生成")
    finally:
        producer.close()
